#include "cache.thread.h"

#include <curl/curl.h>

#include "config-fnt.h"
#include "misc.h"
#include "dialogs.h"

#include <QtCore/QtCore>

#include <memory>
#include <set>
#include <mutex>
#include <queue>
#include <unordered_map>
#include <condition_variable>

#include "cache.thread.options.h"

CacheThread* CacheThread::_instance{nullptr};

struct DownloadItem;
extern "C" size_t writefn(char* ptr, size_t size, size_t nmemb, DownloadItem* f);

struct CacheFile {

	enum class State {
		Empty=0,
		Downloading,
		Ready,
		Error
	};

	CacheThreadPriv* ctp;
	int refc; //*
	qint64 ts; //*
	State state; //*

	std::vector<char*> data;
	size_t size;

	CacheFile(const CacheFile&) =delete;
	CacheFile& operator=(const CacheFile&) =delete;

	CacheFile(CacheThreadPriv* _ctp):
		ctp{_ctp}, refc{0}, ts{0}, state{State::Empty},
		data{}, size{0}
	{
	}
	~CacheFile() {
		for(auto d: data)
			delete[] d;
	}
	size_t memUsage() const {
		return data.size()*BUFSIZ;
	}

};

struct DownloadItem {
	CURLM* curlm;
	QUrl url;
	CURL* curl;
	int login_ver;

	CacheFile* file;
	size_t pos;

	bool finished;

	DownloadItem(CURLM* _curlm, const QUrl& u, CacheFile* c):
		curlm{_curlm}, url{u}, curl{nullptr}, login_ver{0},
		file{c}, pos{0}, finished{false}
	{
		try {
			curl=curl_easy_init();
			if(!curl)
				throwError("Failed to call init CURL easy handle");
			setOptions();
			auto r=curl_multi_add_handle(curlm, curl);
			if(r!=0)
				throwError(curl_multi_strerror(r));
		} catch(...) {
			if(curl) {
				curl_easy_cleanup(curl);
				curl=nullptr;
			}
			throw;
		}
	}

	void finishFile() {
		auto& bufs=file->data;
		while(pos+BUFSIZ<=bufs.size()*BUFSIZ) {
			auto r=bufs.back();
			delete[] r;
			bufs.pop_back();
		}
		file->size=pos;
		finished=true;
	}
	void setOptions() {
		CURLcode r=curl_easy_setopt(curl, CURLOPT_URL, url.toEncoded().data());
		if(r!=0)
			throwError(curl_easy_strerror(r));

		r=curl_easy_setopt(curl, CURLOPT_PROTOCOLS, long(CURLPROTO_FILE|CURLPROTO_HTTP|CURLPROTO_HTTPS|CURLPROTO_FTP|CURLPROTO_FTPS|CURLPROTO_SCP|CURLPROTO_SFTP));
		if(r!=0)
			throwError(curl_easy_strerror(r));

		r=curl_easy_setopt(curl, CURLOPT_WRITEFUNCTION, writefn);
		if(r!=0)
			throwError(curl_easy_strerror(r));

		r=curl_easy_setopt(curl, CURLOPT_WRITEDATA, this);
		if(r!=0)
			throwError(curl_easy_strerror(r));

		/* XXX no connection reuse.
		 * Uncomment this if segfault happens in ``ssh_getworkingpath``.
		 * It's a bug in CURL.
		 r=curl_easy_setopt(curl, CURLOPT_FRESH_CONNECT, 1);
		 if(r!=0)
		 throwError(curl_easy_strerror(r));
		 */
	}
	void resetOptions() {
		auto r=curl_multi_remove_handle(curlm, curl);
		if(r!=0)
			throwError(curl_multi_strerror(r));
		curl_easy_reset(curl);
		setOptions();
		r=curl_multi_add_handle(curlm, curl);
		if(r!=0)
			throwError(curl_multi_strerror(r));
	}
	bool destroy();
	~DownloadItem() {
		try {
			destroy();
		} catch(...) {
		}
	}
	void setLoginInfo(const QString& usr, const QString& pwd, int ver) {
		CURLcode r=curl_easy_setopt(curl, CURLOPT_USERNAME, usr.toLocal8Bit().data());
		if(r!=0)
			throwError(curl_easy_strerror(r));
		r=curl_easy_setopt(curl, CURLOPT_PASSWORD, pwd.toLocal8Bit().data());
		if(r!=0)
			throwError(curl_easy_strerror(r));
		login_ver=ver;
	}
	size_t write(const char* ptr, size_t n) noexcept {
		auto& bufs=file->data;
		auto total=n;
		while(n>0) {
			while(pos>=bufs.size()*BUFSIZ) {
				try {
					auto buf=new char[BUFSIZ];
					file->data.push_back(buf);
				} catch(...) {
					return total-n;
				}
			}
			auto rem=pos%BUFSIZ;
			auto p=bufs[pos/BUFSIZ]+rem;
			auto l=BUFSIZ-rem;
			if(l>n)
				l=n;
			std::char_traits<char>::copy(p, ptr, l);
			ptr+=l;
			n-=l;
			pos+=l;
		}
		return total-n;
	}
};

size_t writefn(char* ptr, size_t size, size_t nmemb, DownloadItem* f) {
	size*=nmemb;
	auto n=f->write(ptr, size);
	if(n<size)
		printMessage("Failed to alloc more memory.");
	return n;
}

struct LoginInfo {
	QString usr;
	QString pwd;
	int ver;
};

struct CacheThreadPriv {
	CacheThread* thread;

	std::vector<LoginInfo> que_logins;
	std::unordered_map<QString, LoginInfo> logins;

	std::mutex mtx_logins;
	std::mutex mtx_files;
	std::mutex mtx_input;
	std::mutex mtx_configs;
	std::condition_variable cv_input;
	std::condition_variable cv_output;
	std::condition_variable cv_logins;

	bool stopRequested;
	std::deque<QUrl> queue_wait;
	std::deque<QUrl> queue_nowait;

	std::unordered_map<QString, CacheFile*> files;
	int64_t tot_size;
	int64_t max_size;

	size_t job_size_wait;
	size_t job_size_nowait;

	void setLoginInfo(DownloadItem* item, const QUrl& url) {
		auto url_root=url.toString(QUrl::RemovePassword|QUrl::RemoveUserInfo|QUrl::RemovePath|QUrl::NormalizePathSegments);
		std::unique_lock<std::mutex> lck{mtx_logins};
		auto i=logins.find(url_root);
		if(i==logins.end()) {
			auto usr=url.userName();
			auto pwd=url.password();
			logins[url_root]=LoginInfo{usr, pwd, 0};
			item->setLoginInfo(usr, pwd, 0);
		} else {
			item->setLoginInfo(i->second.usr, i->second.pwd, i->second.ver);
		}
	}

	CacheFileRef download(const QUrl& url) {
		auto key=url.toString(QUrl::RemovePassword|QUrl::RemoveUserInfo|QUrl::NormalizePathSegments);

		CacheFileRef ret{};
		{
			std::unique_lock<std::mutex> lck{mtx_files};
			auto i=files.find(key);
			if(i!=files.end()) {
				auto file=i->second;
				file->ts=QDateTime::currentDateTime().toMSecsSinceEpoch();
				file->refc++;
				if(file->state==CacheFile::State::Ready)
					return CacheFileRef{file};
				ret=CacheFileRef{file};
			} else {
				auto file=new CacheFile{this};
				file->ts=QDateTime::currentDateTime().toMSecsSinceEpoch();
				file->refc++;
				files[key]=file;
				tot_size+=file->memUsage();
				ret=CacheFileRef{file};
			}
		}

		{
			std::unique_lock<std::mutex> lck{mtx_input};
			queue_wait.push_back(url);
			cv_input.notify_one();
		}
		return ret;
	}
	void downloadIfIdle(const QUrl& url) {
		auto key=url.toString(QUrl::RemovePassword|QUrl::RemoveUserInfo|QUrl::NormalizePathSegments);
		{
			std::unique_lock<std::mutex> lck{mtx_files};
			auto i=files.find(key);
			if(i!=files.end()) {
				auto file=i->second;
				file->ts=QDateTime::currentDateTime().toMSecsSinceEpoch();
				if(file->state==CacheFile::State::Ready)
					return;
			} else {
				auto file=new CacheFile{this};
				file->ts=QDateTime::currentDateTime().toMSecsSinceEpoch();
				files[key]=file;
				tot_size+=file->memUsage();
			}
		}

		{
			std::unique_lock<std::mutex> lck{mtx_input};
			queue_nowait.push_back(url);
			cv_input.notify_one();
		}
	}

	void stop() {
		std::unique_lock<std::mutex> lck{mtx_input};
		stopRequested=true;
		cv_input.notify_one();
	}

	CacheThreadPriv(CacheThread* t):
		thread{t},
		stopRequested{false},
		tot_size{0}, max_size{0}, job_size_wait{0}, job_size_nowait{0}
	{
		auto r=curl_global_init(CURL_GLOBAL_DEFAULT);
		if(r!=0)
			throwError(curl_easy_strerror(r));
	}
	~CacheThreadPriv() {
		while(!files.empty()) {
			auto i=files.begin();
			delete i->second;
			files.erase(i);
		}
		curl_global_cleanup();
	}

	bool multi_wait(CURLM* m) {
		long t=-1;
		CURLMcode r=curl_multi_timeout(m, &t);
		if(r!=0)
			throwError(curl_multi_strerror(r));
		struct timeval timeout{1, 0};
		if(t>=0) {
			timeout.tv_sec=t/1000;
			if(timeout.tv_sec>1)
				timeout.tv_sec=1;
			else
				timeout.tv_usec=(t%1000)*1000;
		}
		fd_set fdread;
		fd_set fdwrite;
		fd_set fdexcep;
		FD_ZERO(&fdread);
		FD_ZERO(&fdwrite);
		FD_ZERO(&fdexcep);
		int maxfd=-1;
		r=curl_multi_fdset(m, &fdread, &fdwrite, &fdexcep, &maxfd);
		if(r!=0)
			throwError(curl_multi_strerror(r));
		if(maxfd==-1)
			return false;
		int rc=select(maxfd+1, &fdread, &fdwrite, &fdexcep, &timeout);
		if(rc==-1)
			throwError(strerror(errno));
		return rc;
	}
	size_t multi_perform(CURLM* m) {
		int n;
		CURLMcode r=curl_multi_perform(m, &n);
		if(r!=0)
			throwError(curl_multi_strerror(r));
		return n;
	}
	void handle_login_denied(ptrdiff_t pi, std::vector<std::unique_ptr<DownloadItem>>& items) {
		auto p=items.begin()+pi;
		auto item=(*p).get();
		auto url=item->url.toString(QUrl::RemovePassword|QUrl::RemoveUserInfo|QUrl::RemovePath|QUrl::NormalizePathSegments);
		std::unique_lock<std::mutex> lck_logins{mtx_logins};
		auto i=logins.find(url);
		if(i==logins.end())
			throwError("Assertion failed");
		auto login_ver=i->second.ver;
		if(item->login_ver!=login_ver) {
			item->resetOptions();
			item->setLoginInfo(i->second.usr, i->second.pwd, login_ver);
			item->pos=0;
			printMessage("Reset password: ", url);
		} else {
			Q_EMIT thread->requestLogin(url, i->second.usr, i->second.pwd);
			while(que_logins.empty())
				cv_logins.wait(lck_logins);
			auto& login=que_logins.front();
			if(login.ver==0) {
				auto url=item->url.toString();
				printMessage("Abort downloading: ", url);
				auto locked=item->destroy();
				items.erase(p);
				if(locked)
					cv_output.notify_all();
			} else {
				i->second.usr=login.usr;
				i->second.pwd=login.pwd;
				i->second.ver+=login.ver;
				item->resetOptions();
				item->setLoginInfo(i->second.usr, i->second.pwd, i->second.ver);
				item->pos=0;
				printMessage("Set new password: ", url);
			}
			que_logins.clear();
		}
	}
	void multi_handle_finished(CURLM* curlm, std::vector<std::unique_ptr<DownloadItem>>& items) {
		int msgs_left;
		while(auto msg=curl_multi_info_read(curlm, &msgs_left)) {
			if(msg->msg==CURLMSG_DONE) {
				auto p=items.end();
				for(auto i=items.begin(); i!=items.end(); i++) {
					if((*i)->curl==msg->easy_handle) {
						p=i;
						break;
					}
				}
				if(p==items.end())
					throwError("Unexpected missing item");
				printMessage("msg.data.result: ", msg->data.result);
				switch(msg->data.result) {
					case CURLE_OK:
						{
							auto item=(*p).get();
							long httpCode=0;
							if(curl_easy_getinfo(item->curl, CURLINFO_RESPONSE_CODE, &httpCode)==CURLE_OK && httpCode==401) {
								handle_login_denied(p-items.begin(), items);
							} else {
								item->finishFile();
								auto url=item->url.toString();
								auto locked=item->destroy();
								items.erase(p);
								if(locked)
									cv_output.notify_all();
							}
						}
						break;
					case CURLE_LOGIN_DENIED:
						handle_login_denied(p-items.begin(), items);
						break;
						/*
							case TIMEOUT:
						// reset
						// redo
						break;
						*/
					default:
						{
							auto item=(*p).get();
							auto url=item->url.toString();
							printMessage("Error downloading: ", url, ": ", curl_easy_strerror(msg->data.result));
							auto locked=item->destroy();
							items.erase(p);
							if(locked>0)
								cv_output.notify_all();
						}
				}
			}
		}
	}

	void tryMakeSpace() {
		int64_t maxSize;
		{
			std::unique_lock<std::mutex> lck{mtx_configs};
			maxSize=max_size;
		}
		std::unique_lock<std::mutex> lck{mtx_files};
		if(tot_size>maxSize) {
			typedef decltype(files.begin()) files_iter;
			std::vector<files_iter> iters{};
			for(auto i=files.begin(); i!=files.end(); i++) {
				auto c=i->second->refc;
				if(c<0)
					throwError("negative refc???");
				if(c==0)
					iters.push_back(i);
			}
			std::sort(iters.begin(), iters.end(), [](const files_iter& a, const files_iter& b) { return a->second->ts>b->second->ts; });
			while(tot_size>maxSize && !iters.empty()) {
				auto i=iters.back();
				iters.pop_back();
				auto f=i->second;
				files.erase(i);
				tot_size-=f->memUsage();
				delete f;
			}
		}
	}

	void run() {
		std::vector<std::unique_ptr<DownloadItem>> items;

		auto curlm_deleter=[](CURLM* m) { auto r=curl_multi_cleanup(m); if(r!=0) throwError(curl_multi_strerror(r)); };
		std::unique_ptr<CURLM, std::function<void(CURLM*)>> curlm{curl_multi_init(), curlm_deleter};
		if(!curlm.get())
			throwError("Failed to init CURL multi handle");

		std::unique_lock<std::mutex> lck_input{mtx_input};
		while(!stopRequested) {
			size_t nWait, nNoWait;
			{
				std::unique_lock<std::mutex> lck{mtx_configs};
				nWait=job_size_wait;
				nNoWait=job_size_nowait;
			}

			{
				std::vector<std::unique_ptr<DownloadItem>> itemsWait;
				std::vector<std::unique_ptr<DownloadItem>> itemsNoWait;
				std::unique_lock<std::mutex> lck_files{mtx_files};
				while(!items.empty()) {
					auto& i=items.back();
					if(i->file->refc>1) {
						itemsWait.push_back(std::move(i));
					} else {
						itemsNoWait.push_back(std::move(i));
					}
					items.pop_back();
				}
				std::swap(items, itemsWait);

				while(!queue_wait.empty() && items.size()<nWait) {
					auto url=queue_wait.front();
					queue_wait.pop_front();
					auto key=url.toString(QUrl::RemovePassword|QUrl::RemoveUserInfo|QUrl::NormalizePathSegments);
					auto i=files.find(key);
					if(i!=files.end()) {
						auto f=i->second;
						if(f->refc>0) {
							switch(f->state) {
								case CacheFile::State::Empty:
								case CacheFile::State::Error:
									{
										std::unique_ptr<DownloadItem> item{new DownloadItem(curlm.get(), key, f)};
										if(!item)
											throwError("Failed to create download item.");
										tot_size-=f->memUsage();
										f->state=CacheFile::State::Downloading;
										f->refc++;
										setLoginInfo(item.get(), url);
										items.push_back(std::move(item));
									}
									break;
								case CacheFile::State::Downloading:
									break;
								case CacheFile::State::Ready:
									if(f->refc>0)
										cv_output.notify_all();
									break;
							}
						}
					}
				}

				if(items.empty()) {
					while(!itemsNoWait.empty()) {
						items.push_back(std::move(itemsNoWait.back()));
						itemsNoWait.pop_back();
					}

					while(!queue_nowait.empty() && items.size()<nNoWait) {
						auto url=queue_nowait.back();
						queue_nowait.pop_back();
						auto key=url.toString(QUrl::RemovePassword|QUrl::RemoveUserInfo|QUrl::NormalizePathSegments);
						auto i=files.find(key);
						if(i!=files.end()) {
							auto f=i->second;
							switch(f->state) {
								case CacheFile::State::Empty:
								case CacheFile::State::Error:
									{
										std::unique_ptr<DownloadItem> item{new DownloadItem(curlm.get(), url, f)};
										if(!item)
											throwError("Failed to create download item.");
										tot_size-=f->memUsage();
										f->state=CacheFile::State::Downloading;
										f->refc++;
										setLoginInfo(item.get(), url);
										printMessage("Start pre-downloading ", key);
										items.push_back(std::move(item));
									}
									break;
								case CacheFile::State::Downloading:
									break;
								case CacheFile::State::Ready:
									break;
							}
						}
					}
				}

				while(!itemsNoWait.empty()) {
					auto i=std::move(itemsNoWait.back());
					itemsNoWait.pop_back();
					i->destroy();
				}
			}
			lck_input.unlock();

			tryMakeSpace();

			if(!items.empty()) {
				auto nleft=multi_perform(curlm.get());
				while(nleft==items.size()) {
					auto r=multi_wait(curlm.get());
					if(r) {
						nleft=multi_perform(curlm.get());
					} else {
						break;
					}
				}
				if(nleft<items.size()) {
					multi_handle_finished(curlm.get(), items);
				}
				lck_input.lock();
				continue;
			}

			lck_input.lock();
			cv_input.wait(lck_input);
		}
	}

	size_t waitForFile(const CacheFileRef& fileToWait) {
		if(!fileToWait)
			return SIZE_MAX;
		std::unique_lock<std::mutex> lck{mtx_files};
		while(true) {
			if(fileToWait->refc<=0)
				throwError("Waiting for an unlocked cache file.");
			if(fileToWait->state==CacheFile::State::Ready||fileToWait->state==CacheFile::State::Error)
				return 0;
			cv_output.wait(lck);
		}
	}
	size_t waitForFiles(const std::vector<CacheFileRef>& filesToWait) {
		bool hasValid=false;
		for(size_t i=0; i<filesToWait.size(); i++) {
			auto& f=filesToWait[i];
			if(f)
				hasValid=true;
		}
		if(!hasValid)
			return SIZE_MAX;

		std::unique_lock<std::mutex> lck{mtx_files};
		while(true) {
			for(size_t i=0; i<filesToWait.size(); i++) {
				auto& f=filesToWait[i];
				if(f) {
					if(f->refc<=0)
						throwError("Waiting for an unlocked cache file.");
					if(f->state==CacheFile::State::Ready||f->state==CacheFile::State::Error)
						return i;
				}
			}
			cv_output.wait(lck);
		}
	}

	void updateLogin(const QString& usr, const QString& pwd, bool cancelled) {
		int verinc=cancelled?0:1;
		std::unique_lock<std::mutex> lck{mtx_logins};
		que_logins.push_back(LoginInfo{usr, pwd, verinc});
		cv_logins.notify_one();
	}
};


// CacheFileRef
CacheFileRef::CacheFileRef(CacheFile* f):
	_file{f}
{
}
void CacheFileRef::release() {
	if(_file) {
		std::unique_lock<std::mutex> lck{_file->ctp->mtx_files};
		_file->refc--;
		_file=nullptr;
	}
}
bool CacheFileRef::isReady() const {
	return _file->state==CacheFile::State::Ready;
}
const std::vector<char*>* CacheFileRef::buffers() const {
	return &_file->data;
}
size_t CacheFileRef::size() const {
	return _file->size;
}

bool DownloadItem::destroy() {
	if(curl) {
		auto r=curl_multi_remove_handle(curlm, curl);
		if(r!=0)
			throwError(curl_multi_strerror(r));
		curl_easy_cleanup(curl);
		curl=nullptr;
	}
	bool locked=false;
	if(file) {
		std::unique_lock<std::mutex> lck{file->ctp->mtx_files};
		file->ctp->tot_size+=file->memUsage();
		file->refc--;
		locked=file->refc>0;
		if(finished) {
			file->state=CacheFile::State::Ready;
		} else {
			file->state=CacheFile::State::Error;
		}
		file=nullptr;
	}
	return locked;
}


// CacheThread
void CacheThread::run() {
	printMessage("Cache thread started");
	try {
		priv->run();
	} catch(const std::exception& e) {
		Q_EMIT threadError(QString{"Unexpected error in cache thread: "}+=e.what());
		printMessage("Cache thread error");
	}
	printMessage("Cache thread finished");
}

CacheThread::CacheThread():
	QThread{Tracer::instance()}, priv{new CacheThreadPriv{this}}
{
	CacheThreadOptions::instance()->attach(priv);
}
CacheThread::~CacheThread() {
	CacheThreadOptions::instance()->attach(nullptr);
	delete priv;
}

void CacheThread::stop() {
	priv->stop();
}

CacheFileRef CacheThread::download(const QUrl& url) {
	if(!url.isValid())
		throwError("URL is not valid.");
	return priv->download(url);
}
void CacheThread::downloadIfIdle(const QUrl& url) {
	if(!url.isValid())
		throwError("URL is not valid.");
	return priv->downloadIfIdle(url);
}

size_t CacheThread::waitForFile(const CacheFileRef& fileToWait) const {
	return priv->waitForFile(fileToWait);
}
size_t CacheThread::waitForFiles(const std::vector<CacheFileRef>& filesToWait) const {
	return priv->waitForFiles(filesToWait);
}

void CacheThread::updateLogin(const QString& usr, const QString& pwd, bool cancelled) {
	return priv->updateLogin(usr, pwd, cancelled);
}

#include "cache.thread.options.cc"
